package com.q.springboot.framework.config.redisson;

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
import org.springframework.core.annotation.MergedAnnotation;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.core.type.filter.AnnotationTypeFilter;
import org.springframework.core.type.filter.AssignableTypeFilter;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;

import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class RMessageListener implements ApplicationListener<ApplicationStartedEvent> {

    /**
     * consumer monitoringMethod monitorMessage
     */
    private final static String METHOD_MONITOR_MESSAGE = "monitorMessage";

    /**
     * redisson topic name
     */
    private final static String ATTRIBUTE_NAME_TOPIC = "topic";

    /**
     * redisson messageQueue name
     */
    private final static String ATTRIBUTE_NAME_QUEUE = "queue";

    /**
     * redisson queue map
     */
    public static Map<String, RBlockingDeque<? super Serializable>> messageQueue = new ConcurrentHashMap<>();

    /**
     * redisson offQueue map
     */
    public static Map<String, RDelayedQueue<? super Serializable>> offQueue = new ConcurrentHashMap<>();

    /**
     * redisson topic map
     */
    public static Map<String, RTopic> topicMap = new ConcurrentHashMap<>();

    @Override
    public void onApplicationEvent(ApplicationStartedEvent applicationStartedEvent) {
        ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false);
        provider.addIncludeFilter(new AnnotationTypeFilter(RMessage.class));
        String basePackage = applicationStartedEvent.getSpringApplication().getMainApplicationClass().getPackage().getName();

        Set<BeanDefinition> beanDefinitions = provider.findCandidateComponents(basePackage);

        ConfigurableListableBeanFactory beanFactory = applicationStartedEvent.getApplicationContext().getBeanFactory();

        mqInit(beanDefinitions, beanFactory);

        provider.clearCache();
        provider.resetFilters(false);

        provider.addIncludeFilter(new AssignableTypeFilter(RMessageConsumer.class));

        Set<BeanDefinition> consumers = provider.findCandidateComponents(basePackage);

        consumerSubscribe(beanFactory, consumers);
    }

    /**
     * consumer subscription news
     *
     * @param beanFactory
     * @param consumers
     */
    private void consumerSubscribe(ConfigurableListableBeanFactory beanFactory, Set<BeanDefinition> consumers) {
        consumers.forEach(beanDefinition -> {
            log.info("rMessage init consumer {}", beanDefinition.getBeanClassName());
            try {
                Object bean = beanFactory.getBean(Class.forName(beanDefinition.getBeanClassName()));
                Method method = bean.getClass().getMethod(METHOD_MONITOR_MESSAGE);
                ReflectionUtils.invokeMethod(method, bean);
            } catch (ClassNotFoundException | NoSuchMethodException e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * Parameter initialization
     *
     * @param beanDefinitions
     * @param beanFactory
     */
    private void mqInit(Set<BeanDefinition> beanDefinitions, final ConfigurableListableBeanFactory beanFactory) {

        RedissonClient redissonClient = beanFactory.getBean(RedissonClient.class);

        beanDefinitions.stream().filter(beanDefinition -> beanDefinition instanceof AnnotatedBeanDefinition).forEach(beanDefinition -> {
            AnnotatedBeanDefinition annotatedBeanDefinition = (AnnotatedBeanDefinition) beanDefinition;
            AnnotationMetadata annotationMetadata = annotatedBeanDefinition.getMetadata();
            MergedAnnotation<RMessage> mergedAnnotation = annotationMetadata.getAnnotations().get(RMessage.class);
            String queryName = mergedAnnotation.getString(ATTRIBUTE_NAME_QUEUE);
            String topicName = mergedAnnotation.getString(ATTRIBUTE_NAME_TOPIC);
            String shortName = topicName + "." + queryName;

            RBlockingDeque<? super Serializable> blockingDeque = redissonClient.getBlockingDeque(shortName);
            messageQueue.put(shortName, blockingDeque);

            RDelayedQueue<? super Serializable> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            offQueue.put(shortName, delayedQueue);

            RTopic topic = redissonClient.getTopic(topicName);
            topicMap.put(shortName, topic);
        });

    }


}
